package natsio

import (
	"encoding/json"
	"errors"
	"gitee.com/liukexing/mqbroker/core/driver/logger"
	"gitee.com/liukexing/mqbroker/core/typedef"
	"github.com/nats-io/go-nats"
	"strconv"
	"sync"
	"time"
)

type NatsPubSub struct {
	typedef.PubSubServer
	Publisher  sync.Map
	Subscriber sync.Map
	DelayMap   sync.Map
	ReplyFunc  sync.Map
	status     bool
}

func (N *NatsPubSub) GetPubConfig() interface{} {
	for _, v := range nodeConfig {
		return v
	}
	return nil
}

func (N *NatsPubSub) GetSubConfig() interface{} {
	for _, v := range nodeConfig {
		return v
	}
	return nil
}

func (N *NatsPubSub) NewPublisher(name string, config interface{}, topic string) bool {
	if _, ok := N.Publisher.Load(name); ok {
		logger.Error(typedef.ErrInfo{
			Info: Config.Info,
			Err:  errors.New("Publisher has exist"),
		})
		return true
	}
	nsqconfig := config.(typedef.NatsConfig)
	producer, err := nats.Connect("nats://" + nsqconfig.HostName + ":" + strconv.Itoa(nsqconfig.Tcp))
	if logger.CheckErr(typedef.ErrInfo{
		Info: Config.Info,
		Err:  err,
	}) {
		return true
	}
	N.Publisher.Store(name, &typedef.NatsPub{
		NatsConn: producer,
		Topic:    topic,
	})
	return false
}

func (N *NatsPubSub) Publish(name string, message string) bool {
	if v, ok := N.Publisher.Load(name); !ok || v == nil {
		logger.Error(typedef.ErrInfo{
			Info: Config.Info,
			Err:  errors.New("Publisher Not Found"),
		})
		return true
	} else {
		pub := v.(*typedef.NatsPub)
		if pub.NatsConn != nil {
			if message == "" {
				logger.Warn(typedef.CommInfo{
					Info:    Config.Info,
					Message: "Message is Empty",
				})
				return true
			}
			return logger.CheckErr(typedef.ErrInfo{
				Info: Config.Info,
				Err:  pub.NatsConn.Publish(pub.Topic, []byte(message)),
			})
		}
	}
	return false
}

func (N *NatsPubSub) DeferredPublish(name string, message string, delay time.Duration) bool {
	if !N.status {
		go N.DelayPubInfo()
	}
	if message == "" {
		logger.Warn(typedef.CommInfo{
			Info:    Config.Info,
			Message: "Message is Empty",
		})
		return true
	}
	N.DelayMap.Store(typedef.GetUUID(), typedef.DelayPub{
		Topic:   name,
		Message: message,
		PubTime: time.Now().Add(delay),
	})
	return false
}

func (N *NatsPubSub) DelPublisher(name string) bool {
	if v, ok := N.Publisher.Load(name); !ok {
		logger.Error(typedef.ErrInfo{
			Info: Config.Info,
			Err:  errors.New("Publisher Not Found"),
		})
		return true
	} else {
		pub := v.(*typedef.NatsPub)
		if pub.NatsConn != nil {
			pub.NatsConn.Flush()
			pub.NatsConn.Close()
			N.Publisher.Delete(name)
		}
	}
	return false
}

func (N *NatsPubSub) NewSubscriber(name string, config interface{}, topic string) bool {
	if _, ok := N.Subscriber.Load(name); ok {
		logger.Error(typedef.ErrInfo{
			Info: Config.Info,
			Err:  errors.New("Subscriber has exist"),
		})
		return true
	}
	natsconfig := config.(typedef.NatsConfig)
	conn, err := nats.Connect("nats://" + natsconfig.HostName + ":" + strconv.Itoa(natsconfig.Tcp))
	if logger.CheckErr(typedef.ErrInfo{
		Info: Config.Info,
		Err:  err,
	}) {
		return true
	}
	N.Subscriber.Store(name, &typedef.NatsSub{
		Topic: topic,
		Conn:  conn,
	})
	return false
}

func (N *NatsPubSub) AddSubscriber(channel string, name string, handel typedef.MessageHandle) bool {
	if v, ok := N.Subscriber.Load(name); !ok {
		logger.Error(typedef.ErrInfo{
			Info: Config.Info,
			Err:  errors.New("Subscriber Not Found"),
		})
		return true
	} else {
		sub := v.(*typedef.NatsSub)
		handle := &NatsHandle{handel: handel}
		subscription, err := sub.Conn.Subscribe(sub.Topic, handle.HandleMessage)
		if logger.CheckErr(typedef.ErrInfo{
			Info: Config.Info,
			Err:  err,
		}) {
			return true
		}
		sub.Subscription.Store(channel, subscription)
	}
	return false
}
func (N *NatsPubSub) RemoveSubscriber(channel string, name string) bool {
	if v, ok := N.Subscriber.Load(name); !ok {
		logger.Error(typedef.ErrInfo{
			Info: Config.Info,
			Err:  errors.New("Subscriber Not Found"),
		})
		return true
	} else {
		sub := v.(*typedef.NatsSub)
		if c, ok := sub.Subscription.Load(channel); !ok {
			logger.Error(typedef.ErrInfo{
				Info: Config.Info,
				Err:  errors.New("Channel Not Found"),
			})
			return true
		} else {
			Channel := c.(*nats.Subscription)
			if Channel != nil {
				sub.Conn.Flush()
				Channel.Unsubscribe()
			}
			sub.Subscription.Delete(channel)
		}
	}
	return false
}
func (N *NatsPubSub) DelSubscriber(name string) bool {
	if v, ok := N.Subscriber.Load(name); !ok {
		logger.Error(typedef.ErrInfo{
			Info: Config.Info,
			Err:  errors.New("Subscriber Not Found"),
		})
		return true
	} else {
		sub := v.(*typedef.NatsSub)
		cnt := 0
		sub.Subscription.Range(func(key, value interface{}) bool {
			cnt++
			return false
		})
		if cnt > 0 {
			logger.Error(typedef.ErrInfo{
				Info: Config.Info,
				Err:  errors.New("Nats Conn Not All Removed"),
			})
			return true
		}
		sub.Conn.Flush()
		sub.Conn.Close()
		N.Subscriber.Delete(name)
	}
	return false
}

func (N *NatsPubSub) Request(name string, message interface{}, timeout time.Duration) typedef.ResponseData {
	if _, ok := N.Publisher.Load("Request"); !ok {
		if N.NewPublisher("Request", N.GetPubConfig(), "Request") {
			return typedef.ResponseData{
				Error:   true,
				Message: "New ReqRsp Failed",
			}
		}
	}
	v, _ := N.Publisher.Load("Request")
	pub := v.(*typedef.NatsPub)
	data := typedef.RequestData{
		Topic: name,
		Data:  message,
		ReqID: typedef.GetUUID(),
	}
	jsv, err := json.Marshal(data)
	if logger.CheckErr(typedef.ErrInfo{
		Info: Config.Info,
		Err:  err,
	}) {
		return typedef.ResponseData{
			Error:   true,
			Message: "Json Marshal Failed",
		}
	}
	msg, err := pub.NatsConn.Request(pub.Topic, jsv, timeout)
	if logger.CheckErr(typedef.ErrInfo{
		Info: Config.Info,
		Err:  err,
	}) {
		return typedef.ResponseData{
			Error:   true,
			Message: "Request Error:" + err.Error(),
		}
	}
	rsp := typedef.ReplyData{}
	if logger.CheckErr(typedef.ErrInfo{
		Info: Config.Info,
		Err:  json.Unmarshal(msg.Data, &rsp),
	}) {
		return typedef.ResponseData{
			Error:   true,
			Message: "Json Unmarshal Failed",
		}
	}
	return typedef.ResponseData{
		Error: false,
		Data:  rsp.Data,
	}
}
func (N *NatsPubSub) AddReply(name string, handel typedef.ReplyHandle) bool {
	if _, ok := N.Subscriber.Load("Request"); !ok {
		if N.NewSubscriber("Request", N.GetSubConfig(), "Request") {
			return true
		}
	}
	v, _ := N.Subscriber.Load("Request")
	sub := v.(*typedef.NatsSub)
	if _, ok := sub.Subscription.Load("Handle"); !ok {
		handle := &NatsReplyHandle{handel: &ReplyHandle{parent: N}}
		subscription, err := sub.Conn.Subscribe(sub.Topic, handle.HandleMessage)
		if logger.CheckErr(typedef.ErrInfo{
			Info: Config.Info,
			Err:  err,
		}) {
			return true
		}
		sub.Subscription.Store("Handle", subscription)
	}

	if _, ok := N.ReplyFunc.Load(name); ok {
		logger.Error(typedef.ErrInfo{
			Info: Config.Info,
			Err:  errors.New("ReplyFunc Has Exist"),
		})
		return true
	}
	N.ReplyFunc.Store(name, handel)
	return false
}
func (N *NatsPubSub) RemoveReply(name string) bool {
	if _, ok := N.ReplyFunc.Load(name); !ok {
		logger.Error(typedef.ErrInfo{
			Info: Config.Info,
			Err:  errors.New("ReplyFunc Not Found"),
		})
		return true
	}
	N.ReplyFunc.Delete(name)
	return false
}

func (N *NatsPubSub) DelayPubInfo() {
	for {
		N.status = true
		N.DelayMap.Range(func(k, value interface{}) bool {
			v := value.(typedef.DelayPub)
			if time.Now().Sub(v.PubTime) > 0 {
				N.Publish(v.Topic, v.Message)
				N.DelayMap.Delete(k)
			}
			return true
		})
		time.Sleep(time.Second)
	}
	defer func() {
		N.status = false
	}()

}
